package com.xuelang.pipeline.worker.app.pod.impl;

import com.xuelang.pipeline.worker.app.common.util.K8sUtils;
import com.xuelang.pipeline.worker.app.common.util.VerifyUtil;
import com.xuelang.pipeline.worker.app.pod.PodService;
import com.xuelang.pipeline.worker.client.pod.dto.PodQueryDTO;
import com.xuelang.pipeline.worker.client.pod.dto.QueryDTO;
import io.kubernetes.client.PodLogs;
import io.kubernetes.client.openapi.ApiException;
import io.kubernetes.client.openapi.apis.CoreV1Api;
import io.kubernetes.client.openapi.models.V1Pod;
import io.kubernetes.client.openapi.models.V1PodList;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;

/**
 * @author liuyuan.ly
 * @since 2024/5/29 9:48
 */
@Service
@Slf4j
public class PodServiceImpl implements PodService {
    @Override
    public V1PodList getPodList(QueryDTO queryDTO) {
        K8sUtils.initApiClient(queryDTO.getKubeConfig());
        String filedName = "";
        if (!StringUtils.isEmpty(queryDTO.getPodName())) {
            filedName = "metadata.name=" + queryDTO.getPodName();
        }

        CoreV1Api api = new CoreV1Api();
        V1PodList v1PodList = null;
        try {
            v1PodList = api.listNamespacedPod(queryDTO.getNamespaceName(), null, null, null,
                    filedName, queryDTO.getLabelName(), null, null, null,
                    null, null);
        } catch (ApiException e) {
            log.info("获取Pod列表信息失败:{},堆栈信息:", e.getResponseBody(), e);
            throw new RuntimeException(e.getResponseBody());
        }
        return v1PodList;
    }

    @Override
    public String getPodLog(PodQueryDTO queryDTO) {
        VerifyUtil.verifyStringNotBlank(queryDTO.getPodName(), "pod名字为空");
        K8sUtils.initApiClient(queryDTO.getKubeConfig());
        CoreV1Api api = new CoreV1Api();
        String podLog = null;
        try {
            podLog = api.readNamespacedPodLog(queryDTO.getPodName(), queryDTO.getNamespaceName(),
                    queryDTO.getContainer(), null, null, null,
                    null, null, null, queryDTO.getTailLines()
                    , false);
        } catch (ApiException e) {
            log.info("获取Pod日志失败:{},堆栈信息:", e.getResponseBody(), e);
            throw new RuntimeException(e.getResponseBody());

        }
        return podLog;
    }

    @Override
    public void streamPodLog(PodQueryDTO queryDTO, WebSocketSession session) {
        log.info("[ Streaming Pod Log ]： 开始获取日志流");
        String kubeConfig = queryDTO.getKubeConfig();
        String podName = queryDTO.getPodName();
        String namespace = queryDTO.getNamespaceName();
        String containerName = queryDTO.getContainer();
        Integer sinceSeconds = queryDTO.getSinceSeconds();
        Integer tailLines = queryDTO.getTailLines();
        int maxRetries = 10; // 最大重试次数
        int retryCount = 0;

        try {
            // 初始化 Kubernetes 客户端
            K8sUtils.initApiClient(kubeConfig);
            CoreV1Api api = new CoreV1Api();
            PodLogs podLogs = new PodLogs();

            // 优先检查历史日志文件
            Path logPath = Paths.get("/mnt/logs", podName + ".log");
            if (Files.exists(logPath)) {
                log.info("[ Streaming Pod Log ]： 发现历史日志文件，直接发送");
                sendLogFileToWebSocket(logPath, session);
                return;
            }
            log.info("[ Streaming Pod Log ]： 尝试获取实时日志流");

            // 若文件不存在，尝试获取实时日志流
            while (retryCount < maxRetries) {
                try {
                    V1Pod pod = api.readNamespacedPodStatus(podName, namespace, null);
                    String podPhase = pod.getStatus().getPhase();

                    if (isLogReady(podPhase)) {
                        InputStream inputStream = podLogs.streamNamespacedPodLog(
                                namespace, podName, containerName, sinceSeconds, tailLines, false
                        );
                        sendLogStreamToWebSocket(inputStream, session);
                        return;
                    } else if ("Unknown".equals(podPhase)) {
                        session.sendMessage(new TextMessage("日志执行出错：Pod 状态未知"));
                        return;
                    } else {
                        session.sendMessage(new TextMessage("日志加载中：Pod 状态为 " + podPhase));
                    }

                    // 等待后重试
                    Thread.sleep(5000);
                    retryCount++;
                } catch (ApiException e) {
                    if (e.getCode() == 404) {
                        // Pod 不存在，再次检查文件（防止竞态条件）
                        if (Files.exists(logPath)) {
                            sendLogFileToWebSocket(logPath, session);
                            return;
                        } else {
                            session.sendMessage(new TextMessage("Pod 不存在且无历史日志"));
                            return;
                        }
                    } else {
                        throw e; // 其他 API 异常继续抛出
                    }
                }
            }

            // 重试结束后再次检查文件
            if (Files.exists(logPath)) {
                sendLogFileToWebSocket(logPath, session);
            } else {
                session.sendMessage(new TextMessage("日志获取超时且无历史日志"));
            }
        } catch (ApiException e) {
            handleApiException(e, session);
        } catch (Exception e) {
            log.error("处理日志请求失败: {}", e.getMessage(), e);
            sendErrorMessage(session, "服务器内部错误");
        } finally {
            log.info("[ Streaming Pod Log ]：停止获取日志流");
        }
    }

    /**
     *  判断 Pod 是否已就绪（可提取为独立方法）
     *
     * @param podPhase
     * @return
     */
    private boolean isLogReady(String podPhase) {
        return "Running".equals(podPhase) || "Failed".equals(podPhase) || "Succeeded".equals(podPhase);
    }

    @Override
    public InputStream streamPodLog(PodQueryDTO queryDTO) {
        String kubeConfig = queryDTO.getKubeConfig();
        String podName = queryDTO.getPodName();
        String namespace = queryDTO.getNamespaceName();
        String containerName = queryDTO.getContainer();
        Integer sinceSeconds = queryDTO.getSinceSeconds();
        Integer tailLines = queryDTO.getTailLines();
        try {
            // 初始化 Kubernetes 客户端
            K8sUtils.initApiClient(kubeConfig);

            // 读取 Pod 信息和日志流
            return new PodLogs().streamNamespacedPodLog(namespace, podName, containerName, sinceSeconds, tailLines, false);

        } catch (Exception e) {
            log.error("Error initializing Kubernetes client: " + e.getMessage(), e);
        }
        return null;
    }

    /**
     *  发送日志文件内容到 WebSocket
     */
    private void sendLogFileToWebSocket(Path logPath, WebSocketSession session) {
        try (BufferedReader reader = Files.newBufferedReader(logPath)) {
            String line;
            while ((line = reader.readLine()) != null) {
                session.sendMessage(new TextMessage(line));
            }
        } catch (IOException e) {
            log.error("读取历史日志文件失败: {}", logPath, e);
            sendErrorMessage(session, "历史日志读取失败");
        }
    }

    /**
     * 发送实时日志流到 WebSocket
     *
     * @param inputStream
     * @param session
     */
    private void sendLogStreamToWebSocket(InputStream inputStream, WebSocketSession session) {
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream))) {
            String line;
            while ((line = reader.readLine()) != null) {
                if (session.isOpen()) {
                    session.sendMessage(new TextMessage(line));
                } else {
                    break;
                }
            }
        } catch (IOException e) {
            log.error("Error streaming log to WebSocket: " + e.getMessage(), e);
        }
    }

    /**
     * 统一处理 API 异常
     *
     * @param e
     * @param session
     */
    private void handleApiException(ApiException e, WebSocketSession session) {
        String errorMessage;
        switch (e.getCode()) {
            case 401:
                errorMessage = "认证失败，请检查 kubeConfig";
                break;
            case 403:
                errorMessage = "权限不足，无法访问 Pod";
                break;
            case 404:
                errorMessage = "Pod 不存在";
                break;
            default:
                errorMessage = "Kubernetes API 错误: " + e.getMessage();
        }
        sendErrorMessage(session, errorMessage);
    }

    /**
     * 统一发送错误消息
     *
     * @param session
     * @param message
     */
    private void sendErrorMessage(WebSocketSession session, String message) {
        try {
            session.sendMessage(new TextMessage(message));
        } catch (IOException e) {
            log.error("发送错误消息失败: {}", e.getMessage());
        }
    }

}
